package org.jxkj.app

import java.text.SimpleDateFormat
import java.util.Date
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.storage.StorageLevel
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{when, _}
import org.jxkj.data.IDaoProvider
import org.jxkj.util.JzCondition

import scala.collection.JavaConversions._

/**
 * 1、日发电量考核
 * 2、增发电量考核
 */
object DailyGenerationCheck {
  @Deprecated
  //日发电量考核旧版
  private def calculateData(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider,time: Array[String], step : Int): Unit = {
    import ss.implicits._
    var startTime = time(0)
    val firstTime = time(0)
    val endTime = time(1)
    val endTime2=endTime+" 23:59:59"
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
    val y : Float = 0.02f
    val lower = 49.9
    val upper = 50.1
    val a = 1
    val b = 1

    //var startTimeStamp = sdf.parse(startTime).getTime()
    //var lagDate = sdfs.format(new Date(startTimeStamp-60*15*1000))
    //println(lagDate)
    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_daily_generation_point_conf")
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver2)
      .load().cache()
    jdbcDF.createOrReplaceTempView("hms_daily_generation_point_conf")

    import ss.implicits._

    val point_code: Array[String] = jdbcDF.map(x=>x.getString(0)).collect()

    //val point_code= Array("JKXT_EJ_AI02489","JKXT_EJ_AI02491","JKXT_EJ_AI02651","JKXT_EJ_DI09485","JKXT_EJ_DI09484")
    //for(i <- 0 until  point_code.length){println(point_code(i))}
    //用到15分钟补点
    GetData.getBySecond(ss,sc,hTable,point_code,time,step)

    //ss.sql("select * from t_ods_hbase_edos_tmp where createTime>'2020/04/07 08:30:00' order by createTime,checkPoint").show(500,0)
    //ss.sql("select * from t_ods_hbase_edos").show(500)

    val odsDF = ss.sql("select createTime,sum(case when instr(a.point_name,'原始值')>0 then pvalue end) plan_out" +
      " ,sum(case when  instr(a.point_name,'实际出力')>0 then pvalue end) meter_out" +
      ",sum(case when instr(a.point_name,'采用值')>0 then pvalue end) actual_plan" +
      ",sum(case when  instr(a.point_name,'大江频率')>0 then pvalue end) power_dj" +
      ",sum(case when  instr(a.point_name,'二江频率')>0 then pvalue end) power_erj" +
      " from hms_daily_generation_point_conf  a " +
      " join t_ods_hbase_edos_tmp b on b.checkPoint=a.point_code " +
      " group by createTime" )
    odsDF.createOrReplaceTempView("t_ods")
    //odsDF.orderBy("createTime").show(500)

    val etldf = ss.sql("select createTime ass_time" +
      s",case when abs(actual_plan-meter_out)/actual_plan > $y and power_dj between $lower and $upper and power_erj between $lower and $upper " +
      s" then round(abs(actual_plan-meter_out)*$b*5/60,2) else round(abs(actual_plan-meter_out)*$a*5/60,2) end ass_electric_quantity" +
      ",round(plan_out,2) plan_out,round(meter_out,2) meter_out,round(actual_plan,2) as actual_out" +
      ",round(abs(actual_plan-meter_out)/actual_plan,4) as deviation" +
      ",0 actual_ass_electric_quantity,0 agc,'' remark"+
      ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time" +
        s" from t_ods a where (abs(actual_plan-meter_out)/actual_plan > $y and power_dj between $lower and $upper and power_erj between $lower and $upper)" +
      s" or power_dj<$lower or power_dj>$upper  or power_erj<$lower or power_erj>$upper ")
    //etldf.orderBy("ass_time").show(500)

    val tablename="hms_eo_daily_generation_assess"
    val delString=s"delete from $tablename where ass_time between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString)

    //写mysql
    etldf.write
      .format("jdbc")
      .mode("append")
      .option("url",MysqlDml.url2)
      .option("dbtable",tablename)
      .option("user",MysqlDml.user2)
      .option("password",MysqlDml.password2)
      .option("driver",MysqlDml.driver2)
      .save()


  }

  //增发电量
  @Deprecated
  private def calculateData2(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider,time: Array[String], step : Int): Unit = {
    import ss.implicits._
    var startTime = time(0)
    val endTime = time(1)
    var startTime2 = startTime+" 00:00:00"
    val endTime2=endTime+" 23:59:59"
    val time2=Array(startTime2,endTime2)
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")

    //var startTimeStamp = sdf.parse(startTime).getTime()
    //var lagDate = sdfs.format(new Date(startTimeStamp-60*15*1000))
    //println(lagDate)
    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_first_frequency_point_conf")
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver2)
      .load().cache()
    jdbcDF.createOrReplaceTempView("hms_first_frequency_point_conf")

    val hms_cm_asset = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_cm_asset")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    hms_cm_asset.createOrReplaceTempView("hms_cm_asset")

    import ss.implicits._

    val point_code: Array[String] = jdbcDF.filter($"category" ===  "power").map(x=>x.getString(0)).collect()
    //val point_code =Array("JKXT_EJ_AI03079")
    val table="hms_first_frequency_point_conf"
    GetData.getBySecond(ss,sc,hTable,point_code,time2,step,1)
    //GetData.getAndFill(ss,sc,hTable,point_code,time,step,table)
    //ss.sql("select * from t_ods_hbase_edos_tmp where createTime>'2020/04/07 08:30:00' order by createTime,checkPoint").show(500,0)
    //ss.sql("select * from t_ods_hbase_edos").show(500)

    val odsDF = ss.sql("select substr(point_name,1,3) as  unit_name,createTime,pvalue" +
      " ,case when substr(point_name,1,3)='00F' THEN 20 when substr(point_name,1,3) in ('01F','02F') then 170 else 125 end power_rated " +
      " from hms_first_frequency_point_conf  a " +
      " left join t_ods_hbase_edos_tmp b on b.checkPoint=a.point_code " +
      " where a.category = ('power') and createTime is not null and substr(point_name,1,3) not in ('00F','01F','02F')" )
    odsDF.createOrReplaceTempView("t_ods")
    //odsDF.orderBy("createTime").show(500)

    val etldf = ss.sql("select assetid locationsid,substr(createTime,1,10) stat_time,sum(case when pvalue>=power_rated then pvalue-power_rated else 0 end)/3600 increases_electricity" +
      ",'3' type, case when unit_name<='07F' THEN '二江' else '大江' end project_name"+
      ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time" +
      " from t_ods a left join hms_cm_asset b " +
      "on concat(a.unit_name,'机组')=b.description" +
      " group by assetid,substr(createTime,1,10), case when unit_name<='07F' THEN '二江' else '大江' end")
    //etldf.orderBy("locationsid","stat_time").show(500)

    val etldf2 = ss.sql("select assetid locationsid,substr(createTime,1,13) stat_time,sum(case when pvalue>=power_rated then pvalue-power_rated else 0 end)/3600 increases_electricity" +
      ",'4' type, case when unit_name<='07F' THEN '二江' else '大江' end project_name"+
      ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time" +
      " from t_ods a left join hms_cm_asset b " +
      "on concat(a.unit_name,'机组')=b.description" +
      " group by assetid,substr(createTime,1,13), case when unit_name<='07F' THEN '二江' else '大江' end")

    val tablename="hms_eo_daily_increases_electricity"
    val delString=s"delete from $tablename where substr(stat_time,1,10) between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString)

    //写mysql
    etldf.write
      .format("jdbc")
      .mode("append")
      .option("url",MysqlDml.url2)
      .option("dbtable",tablename)
      .option("user",MysqlDml.user2)
      .option("password",MysqlDml.password2)
      .option("driver",MysqlDml.driver2)
      .save()

    etldf2.write
      .format("jdbc")
      .mode("append")
      .option("url",MysqlDml.url2)
      .option("dbtable",tablename)
      .option("user",MysqlDml.user2)
      .option("password",MysqlDml.password2)
      .option("driver",MysqlDml.driver2)
      .save()


  }

//日发电量考核新版
  @Deprecated
  private def calculateData3(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider,time: Array[String], step : Int): Unit = {
    import ss.implicits._
    var startTime = time(0)
    val firstTime = time(0)
    val endTime = time(1)
    val endTime2=endTime+" 23:59:59"
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
    val y : Float = 0.02f
    val lower = 49.9
    val upper = 50.1
    val a = 1
    val b = 1

    //var startTimeStamp = sdf.parse(startTime).getTime()
    //var lagDate = sdfs.format(new Date(startTimeStamp-60*15*1000))
    //println(lagDate)
    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_daily_generation_point_conf")
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver2)
      .load().cache()
    jdbcDF.createOrReplaceTempView("hms_daily_generation_point_conf")

    import ss.implicits._

    val point_code: Array[String] = jdbcDF.map(x=>x.getString(0)).collect()

    //val point_code= Array("JKXT_EJ_AI02489","JKXT_EJ_AI02491","JKXT_EJ_AI02651","JKXT_EJ_DI09485","JKXT_EJ_DI09484")
    //for(i <- 0 until  point_code.length){println(point_code(i))}
    //用到15分钟补点
    GetData.getBySecond(ss,sc,hTable,point_code,time,step)

    //ss.sql("select * from t_ods_hbase_edos_tmp where createTime>'2020/04/07 08:30:00' order by createTime,checkPoint").show(500,0)
    //ss.sql("select * from t_ods_hbase_edos").show(500)

    val odsDF = ss.sql("select createTime,sum(case when instr(a.point_name,'原始值')>0 then pvalue end) plan_out" +
      " ,sum(case when  instr(a.point_name,'实际出力')>0 then pvalue end) meter_out" +
      ",sum(case when instr(a.point_name,'采用值')>0 then pvalue end) actual_plan" +
      ",sum(case when  instr(a.point_name,'大江频率')>0 then pvalue end) power_dj" +
      ",sum(case when  instr(a.point_name,'二江频率')>0 then pvalue end) power_erj" +
      " from hms_daily_generation_point_conf  a " +
      " join t_ods_hbase_edos_tmp b on b.checkPoint=a.point_code " +
      " group by createTime" )
    odsDF.createOrReplaceTempView("t_ods")
    //odsDF.orderBy("createTime").show(500)

    val etldf = ss.sql("select createTime ass_time" +
      s",case when power_dj between $lower and $upper and power_erj between $lower and $upper " +
      s" then round(abs(actual_plan-meter_out)*$b/12,2) else round(abs(actual_plan-meter_out)*$a/6,2) end ass_electric_quantity" +
      s",case when power_dj between $lower and $upper and power_erj between $lower and $upper " +
      s" then round((abs(actual_plan-meter_out)-actual_plan*$y)/6,2) else round((abs(actual_plan-meter_out)-actual_plan*$y)/3,2) end ass_electric_quantity_new" +
      ",round(plan_out,2) plan_out,round(actual_plan,2) as actual_plan,round(meter_out,2) meter_out" +
      s",case when power_dj between $lower and $upper and power_erj between $lower and $upper then '正常' else '异常' end freq" +
      ",round(abs(actual_plan-meter_out)/actual_plan,4) as deviation" +
      ",round(abs(plan_out-meter_out)/plan_out,4) as deviation_plan" +
      ",'' remark"+
      ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time" +
      s" from t_ods a where abs(actual_plan-meter_out)-actual_plan*$y > 0" )
    //etldf.orderBy("ass_time").show(500)

    val tablename="hms_eo_daily_generation_c"
    val delString=s"delete from $tablename where ass_time between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString)

    //写mysql
    etldf.write
      .format("jdbc")
      .mode("append")
      .option("url",MysqlDml.url2)
      .option("dbtable",tablename)
      .option("user",MysqlDml.user2)
      .option("password",MysqlDml.password2)
      .option("driver",MysqlDml.driver2)
      .save()


  }

  //日发电量考核最新版 计划出力按斜率补数
  @Deprecated
  private def calculateData4(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider,time: Array[String], step : Int): Unit = {
    import ss.implicits._
    var startTime = time(0)
    val firstTime = time(0)
    val endTime = time(1)
    val endTime2=endTime+" 23:59:59"
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
    val y : Float = 0.02f
    val lower = 49.9
    val upper = 50.1
    val a = 1
    val b = 1

    //var startTimeStamp = sdf.parse(startTime).getTime()
    //var lagDate = sdfs.format(new Date(startTimeStamp-60*15*1000))
    //println(lagDate)
    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_daily_generation_point_conf")
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver2)
      .load().cache()
    jdbcDF.createOrReplaceTempView("hms_daily_generation_point_conf")

    import ss.implicits._
    //GetData.getBySecond(ss,sc,hTable,Array("GZB_TD_AI00001"),time,900)
    GetData.getNoLoop(ss,sc,hTable,Array("GZB_TD_AI00001"),time,step)
    ss.sql("select checkPoint,createTime,pvalue*10 as pvalue from t_ods_hbase_edos").createOrReplaceTempView("t_ods_plan")
    val point_code: Array[String] = jdbcDF.map(x=>x.getString(0)).collect()

    //val point_code= Array("JKXT_EJ_AI02489","JKXT_EJ_AI02491","JKXT_EJ_AI02651","JKXT_EJ_DI09485","JKXT_EJ_DI09484")
    //for(i <- 0 until  point_code.length){println(point_code(i))}
    //用到15分钟补点
    GetData.getBySecond(ss,sc,hTable,point_code,time,step)

    //ss.sql("select * from t_ods_hbase_edos_tmp where createTime>'2020/04/07 08:30:00' order by createTime,checkPoint").show(500,0)
    //ss.sql("select * from t_ods_hbase_edos").show(500)

    val odsDF = ss.sql("select createTime" +
      //",sum(case when instr(a.point_name,'原始值')>0 then pvalue end) plan_out" +
      " ,sum(case when  instr(a.point_name,'实际出力')>0 then pvalue end) meter_out" +
      ",sum(case when instr(a.point_name,'采用值')>0 then pvalue end) actual_plan" +
      ",sum(case when  instr(a.point_name,'大江频率')>0 then pvalue end) power_dj" +
      ",sum(case when  instr(a.point_name,'二江频率')>0 then pvalue end) power_erj" +
      ",max(case when a.point_name='大江网控AGC投入' then pvalue end) agc_dj" +
      ",max(case when a.point_name='二江网控AGC投入' then pvalue end) agc_ej" +
      " from hms_daily_generation_point_conf  a " +
      " join t_ods_hbase_edos_tmp b on b.checkPoint=a.point_code " +
      " group by createTime" )
    odsDF.createOrReplaceTempView("t_ods1")
    //odsDF.orderBy("createTime").show(500)
    ss.sql("select a.createTime,meter_out,b.pvalue as pvalue,actual_plan,power_dj,power_erj,agc_dj,agc_ej" +
      ",case when b.pvalue is null and unix_timestamp(a.createTime,'yyyy/MM/dd HH:mm:ss')%900=300 then lag(b.pvalue,1) over(order by a.createTime) else lag(b.pvalue,2) over(order by a.createTime) end lag_value" +
      ",case when b.pvalue is null and unix_timestamp(a.createTime,'yyyy/MM/dd HH:mm:ss')%900=300 then lead(b.pvalue,2) over(order by a.createTime) else lead(b.pvalue,1) over(order by a.createTime) end lead_value" +
      " from t_ods1 a left join t_ods_plan b on a.createTime=b.createTime" ).createOrReplaceTempView("t_ods2")
    //ss.sql("select * from t_ods2 order by createTime").show(500)

    ss.sql("select createTime,meter_out" +
      ",nvl(pvalue,case when unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss')%900=300 then lag_value+(nvl(lead_value,lag_value)-lag_value)/3 else lag_value+(nvl(lead_value,lag_value)-lag_value)*2/3 end) plan_out" +
      ",actual_plan,power_dj,power_erj,agc_dj,agc_ej from t_ods2").createOrReplaceTempView("t_ods")

    //ss.sql("select * from t_ods order by createTime").show(500)

    val etldf = ss.sql("select createTime ass_time" +
      s",case when power_dj between $lower and $upper and power_erj between $lower and $upper " +
      s" then round((abs(plan_out-meter_out)-plan_out*$y)/6,2) else round(abs(plan_out-meter_out)/6,2) end ass_electric_quantity" +
      ",round(plan_out,2) plan_out,round(actual_plan,2) as actual_plan,round(meter_out,2) meter_out" +
      s",case when power_dj between $lower and $upper and power_erj between $lower and $upper then '正常' else '异常' end freq" +
      ",round(abs(actual_plan-meter_out)/actual_plan,4) as deviation" +
      ",round(abs(plan_out-meter_out)/plan_out,4) as deviation_plan" +
      s",'' remark,concat(nvl(agc_dj,'#'),nvl(agc_ej,'#')) as agc"+
      s",case when abs(plan_out-meter_out)-plan_out*$y > 0 and (agc_dj=0 or agc_ej=0) then '1' else '0' end ass_flag"+
      s",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time" +
      s" from t_ods a" )

    //etldf.orderBy("ass_time").show(500)

    //val tablename="hms_eo_daily_generation_c"
     val tablename="hms_eo_daily_generation_assess"
    val delString=s"delete from $tablename where ass_time between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString)

    //写mysql
    etldf.write
      .format("jdbc")
      .mode("append")
      .option("url",MysqlDml.url2)
      .option("dbtable",tablename)
      .option("user",MysqlDml.user2)
      .option("password",MysqlDml.password2)
      .option("driver",MysqlDml.driver2)
      .save()


  }


  //日发电量考核最新版 /*计划出力按斜率补数*/ 分原始计划和采用计划
  def calcDayElectricQty(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step : Int): Unit = {
    import ss.implicits._
    var startTime = time(0)
    val firstTime = time(0)
    val endTime = time(1)
    val endTime2=endTime//+" 23:59:59"
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")
    val y : Float = 0.02f
    val lower = 49.9
    val upper = 50.1
    val a = 1
    val b = 1
    val c = 10

    //var startTimeStamp = sdf.parse(startTime).getTime()
    //var lagDate = sdfs.format(new Date(startTimeStamp-60*15*1000))
    //println(lagDate)
    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_daily_generation_point_conf")
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver2)
      .load().cache()
    jdbcDF.createOrReplaceTempView("hms_daily_generation_point_conf")

    import ss.implicits._
    val point_code: Array[String] = jdbcDF.map(x=>x.getString(0)).collect()

    //val point_code= Array("JKXT_EJ_AI02489","JKXT_EJ_AI02491","JKXT_EJ_AI02651","JKXT_EJ_DI09485","JKXT_EJ_DI09484")
    //for(i <- 0 until  point_code.length){println(point_code(i))}
    //用到15分钟补点
    GetData.getByWebApiTo(ss,sc,hTable,point_code,time,step,300)

    //ss.sql("select * from t_ods_hbase_edos_tmp where createTime>'2020/04/07 08:30:00' order by createTime,checkPoint").show(500,0)
    //ss.sql("select * from t_ods_hbase_edos").show(500)

//  原始值改测点 现在的测点要 *10 -- * $c
    val odsDF = ss.sql("select createTime" +
      s",sum(case when instr(a.point_name,'原始值')>0 then pvalue end) * $c original_plan" +
      " ,sum(case when  instr(a.point_name,'实际出力')>0 then pvalue end) meter_out" +
      ",sum(case when instr(a.point_name,'采用值')>0 then pvalue end) actual_plan" +
      ",sum(case when  instr(a.point_name,'大江频率')>0 then pvalue end) power_dj" +
      ",sum(case when  instr(a.point_name,'二江频率')>0 then pvalue end) power_erj" +
      ",max(case when a.point_name='大江网控AGC投入' then pvalue end) agc_dj" +
      ",max(case when a.point_name='二江网控AGC投入' then pvalue end) agc_ej" +
      " from hms_daily_generation_point_conf  a " +
      " join t_ods_hbase_edos_tmp b on b.checkPoint=a.point_code " +
      " group by createTime" )
    odsDF.createOrReplaceTempView("t_ods")

//    odsDF.show()

    /*
* ass_time 考核时间
  original_plan 原始计划（MW）
  actual_plan 采用计划（MW）
  meter_out 实际出力（MW）
  freq 电网频率
  deviation_original 实际出力与原始计划偏差（%）
  deviation_actual 实际出力与采用计划偏差（%）
  agc 全厂AGC网控投退情况
  original_deviation_flag 原始计划偏差情况
  actual_deviation_flag 采用计划偏差情况
  original_ass_flag 原始计划偏差考核情况 1考核0不考核
  actual_ass_flag 采用计划偏差考核情况 1考核0不考核
  original_two_ass_flag 原始计划“两个细则”考核情况 1考核0不考核
  actual_two_ass_flag 采用计划“两个细则”考核情况 1考核0不考核
  original_ass_electric_quantity 原始计划考核电量（MW·h）
  actual_ass_electric_quantity 采用计划考核电量（MW·h）
* */
    val etldf = ss.sql("select createTime ass_time" +

      ",round(original_plan,2) original_plan" +
      ",round(actual_plan,2) as actual_plan" +
      ",round(meter_out,2) meter_out" +
      s",case when power_dj between $lower and $upper and power_erj between $lower and $upper then '正常' else '异常' end freq" +
      ",round(abs(original_plan-meter_out)/original_plan,4) as deviation_original" +
      ",round(abs(actual_plan-meter_out)/actual_plan,4) as deviation_actual" +

      ",concat(nvl(agc_dj,'#'),nvl(agc_ej,'#')) as agc"+
      s",case when abs(original_plan-meter_out)-original_plan*$y > 0 then '偏差异常' when abs(original_plan-meter_out)-original_plan*$y <= 0 then '偏差正常' else '数据异常' end original_deviation_flag"+
      s",case when abs(actual_plan-meter_out)-actual_plan*$y > 0 then '偏差异常' when abs(actual_plan-meter_out)-actual_plan*$y <= 0 then '偏差正常' else '数据异常' end actual_deviation_flag"+

      s",case when abs(original_plan-meter_out)-original_plan*$y > 0 then '1' else '0' end original_ass_flag"+
      s",case when abs(actual_plan-meter_out)-actual_plan*$y > 0 then '1' else '0' end actual_ass_flag"+
      s",case when abs(original_plan-meter_out)-original_plan*$y > 0  and (agc_dj=0 or agc_ej=0) then '1' else '0' end original_two_ass_flag"+
      s",case when abs(actual_plan-meter_out)-actual_plan*$y > 0  and (agc_dj=0 or agc_ej=0) then '1' else '0' end actual_two_ass_flag"+

      s",case when power_dj between $lower and $upper and power_erj between $lower and $upper " +
      s" then round((abs(original_plan-meter_out)-original_plan*$y)/6,2) else round(abs(original_plan-meter_out)/6,2) end original_ass_electric_quantity" +
      s",case when power_dj between $lower and $upper and power_erj between $lower and $upper " +
      s" then round((abs(actual_plan-meter_out)-actual_plan*$y)/6,2) else round(abs(actual_plan-meter_out)/6,2) end actual_ass_electric_quantity" +
      ",'' remark" +
      ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time" +
      " from t_ods a" )

    //etldf.orderBy("ass_time").show(500)

    //val tablename="hms_eo_daily_generation_c"
    val tablename="hms_eo_daily_generation_assess"
    val delString=s"delete from $tablename where ass_time between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString)

    //写mysql
    etldf.write
      .format("jdbc")
      .mode("append")
      .option("url",MysqlDml.url2)
      .option("dbtable",tablename)
      .option("user",MysqlDml.user2)
      .option("password",MysqlDml.password2)
      .option("driver",MysqlDml.driver2)
      .save()


  }

  //增发电量考核
  def calcIncreasesElectricity(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String],step:Int): Unit = {

    var startTime = time(0)
    val endTime = time(1)
    var startTime2 = startTime//+" 00:00:00"
    val endTime2=endTime//+" 23:59:59"
    val time2=Array(startTime2,endTime2)
    val a = 60
    val b = 3600

    JzCondition.calData(ss,sc,hTable,time2,step,1)

    val jdbc2DF = ss.read
      .format("jdbc")
      .option("url",MysqlDml.url)
      .option("dbtable","t_unit_electric_point_conf_check")
      .option("user",MysqlDml.user)
      .option("password",MysqlDml.password)
      .option("driver",MysqlDml.driver2)
      .load()
    jdbc2DF.createOrReplaceTempView("t_unit_electric_point_conf")

    val hms_cm_asset = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_cm_asset")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load().cache()
    hms_cm_asset.createOrReplaceTempView("hms_cm_asset")

    val jdbc4DF = ss.read
      .format("jdbc")
      .option("url",MysqlDml.url2)
      .option("dbtable","hms_eo_waterhead_power_n87")
      .option("user",MysqlDml.user)
      .option("password",MysqlDml.password)
      .option("driver",MysqlDml.driver2)
      .load()
    jdbc4DF.createOrReplaceTempView("hms_eo_waterhead_power")

    //生成配置表

    ss.sql("select b.sby1 project_name,substr(b.description,1,3) unit_name,water_head a2,power_n87 b2" +
      ",lag(water_head,1) over(partition by a.locationsid order by water_head) a1" +
      ",lag(power_n87,1) over(partition by a.locationsid order by water_head) b1" +
      " from hms_eo_waterhead_power a left join hms_cm_asset b on a.locationsid=b.assetid")
      .createOrReplaceTempView("t_unit_power_limit_conf_tmp")

    ss.sql("select project_name,unit_name,a2,b2,nvl(a1,0) as a1,nvl(b1,b2) as b1" +
      " from t_unit_power_limit_conf_tmp").createOrReplaceTempView("t_unit_power_limit_conf")

    //ss.sql("select *from t_unit_electric_limit_conf order by unit_name,a2").show()
    //ss.sql("select *from t_unit_power_limit_conf order by unit_name,a2").show()

    val unitDF= ss.sql("select unit_name,point_code_st from t_unit_electric_point_conf " +
      "union all select unit_name,point_code_power from t_unit_electric_point_conf ")
    import ss.implicits._
    val arrayString: Array[String] = unitDF.map{ x=>x.getString(1)}.collect()

    GetData.getBySecond(ss,sc,hTable,arrayString,time2,step,1)
    val resultDf = ss.sql("select a.unit_name,b.createTime,round(b.pvalue,2) pvalue_st,round(f.pvalue,2) pvalue_power,nvl(working_name,'机组稳定运行态') working_name" +
      " from t_unit_electric_point_conf a" +
      "  left join t_ods_hbase_edos_tmp b on a.point_code_st=b.checkPoint" +
      "  left join t_ods_hbase_edos_tmp f on a.point_code_power=f.checkPoint and b.createTime=f.createTime" +
      "  left join t_jzgk_statu e on a.unit_name=e.crew_code and b.createTime=e.start_time").cache()
    resultDf.createOrReplaceTempView("t_unit_electric_st_ly_dy_power")

    //增发电量
    val result2Df =ss.sql("select b.project_name,a.unit_name,createTime,pvalue_st,pvalue_power,working_name, " +
      "(b1+(pvalue_st-a1)*(b2-b1)/(a2-a1)) as power_n87" +
      " from t_unit_electric_st_ly_dy_power a" +
      " left join t_unit_power_limit_conf b on a.pvalue_st >b.a1 and a.pvalue_st<=b.a2" +
      " where instr(b.unit_name,a.unit_name)>0").cache()
    //result2Df.show()
    result2Df.createOrReplaceTempView("t_unit_st_ly_dy_power_range")
//  --2021-03-15 根据要求给是否开机加一个 有功大于60
    ss.sql(s"select project_name,createTime,case when sum(pvalue_power) > $a  and sum(pvalue_power)-sum(power_n87)<=0 then 0 else 1 end iscal_flag" +
      " from t_unit_st_ly_dy_power_range where group by project_name,createTime")
      .createOrReplaceTempView("t_iscal_flag")

    //按小时统计增发电量  加入工况 大江二江总量条件 --- 2021-03-09 改 b.iscal_flag=1 为 b.iscal_flag= 0
    val resultHour=ss.sql("select a.project_name,a.unit_name,substr(a.createTime,1,13) createTime" +
      ",max(case when working_name='机组稳定运行态' then 1 else 0 end) working_name" +
      ",avg(pvalue_st) pvalue_st,avg(pvalue_power) pvalue_power,avg(power_n87) power_n87" +
     // s",sum(case when working_name='机组稳定运行态' and b.iscal_flag= 0 and pvalue_power>power_n87 then (pvalue_power-power_n87)/$b else 0 end) increases_electricity" +
     // s",sum(case when working_name='机组稳定运行态' and pvalue_power>power_n87 then (pvalue_power-power_n87)/$b else 0 end) increases_electricity" +
     // s",sum(case when pvalue_power>power_n87 then (pvalue_power-power_n87)/$b else 0 end) increases_electricity" +
      s",sum(case when working_name='机组稳定运行态' and b.iscal_flag= 0 and pvalue_power>power_n87 then (pvalue_power-power_n87)/$b else 0 end) increases_electricity" +
      ",'4' type"+
      " from t_unit_st_ly_dy_power_range a left join t_iscal_flag b" +
      " on a.project_name=b.project_name and a.createTime=b.createTime" +
      " group by a.project_name,a.unit_name,substr(a.createTime,1,13)")

    resultHour.createOrReplaceTempView("t_increases_electricity")

    val resultDay=ss.sql("select a.project_name,a.unit_name,substr(a.createTime,1,10) createTime,max(working_name) working_name" +
      ",avg(pvalue_st) pvalue_st,avg(pvalue_power) pvalue_power,avg(power_n87) power_n87" +
      ",sum(increases_electricity) increases_electricity" +
     // ",pvalue_power-power_n87 increases_electricity" +
      ",'3' type"+
      " from t_increases_electricity a  " +
      " group by a.project_name,a.unit_name,substr(a.createTime,1,10)")

    resultHour.union(resultDay).createOrReplaceTempView("t_increases_electricity2")

    val result3Df =ss.sql("select project_name,c.assetid as locationsid,createTime as stat_time,working_name condition_id,pvalue_st water_head,pvalue_power actual_power, " +
      //"power_n87,increases_electricity,type,"+
      //"power_n87,pvalue_power-power_n87 increases_electricity,type,"+
      "(case when power_n87 is null then 0 else power_n87 end) power_n87,(case when (pvalue_power-power_n87) is null then 0 else (pvalue_power-power_n87) end) increases_electricity,type,"+
      "'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time" +
      " from t_increases_electricity2 a" +
      "  left join hms_cm_asset c on concat(a.unit_name,'机组')=c.description")
    //result3Df.show()
    //resultDf.repartition(1).write.csv("F://user/root/result.csv")

    //对数据进行操作
    var list = result3Df.collectAsList().listIterator();
    var count = 0.0;
    while( list.hasNext ) {
      //val t = list.next().getAs("increases_electricity").asInstanceOf[Double];
      count += list.next().getAs("increases_electricity").asInstanceOf[Double];
    }

    val tableName2 ="hms_eo_daily_increases_electricity"
    MysqlDml.delete2(s"delete from $tableName2 where stat_time between  '$startTime'  and  '$endTime'")

    if( count > 0.0 ) {
      val resultNew = result3Df.withColumn("status",lit(1) );
      //resultNew.show(1000);

      //写mysql
      resultNew.write
        .format("jdbc")
        .mode("append")
        .option("url", MysqlDml.url2)
        .option("dbtable", tableName2)
        .option("user",MysqlDml.user2)
        .option("password",MysqlDml.password2)
        .option("driver", MysqlDml.driver2)
        .save()
    } else {
      val resultNew = result3Df.withColumn("status",lit(0) );
      //resultNew.show(1000);

      //写mysql
      resultNew.write
        .format("jdbc")
        .mode("append")
        .option("url", MysqlDml.url2)
        .option("dbtable", tableName2)
        .option("user",MysqlDml.user2)
        .option("password",MysqlDml.password2)
        .option("driver", MysqlDml.driver2)
        .save()
    }

  }
}
